package com.blank.shop.ssr.runner;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import io.rebloom.client.Client;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.data.geo.*;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * <br/>Date 2021/9/3
 * <br/>Time 12:32:06
 *
 * @author _blank
 */
@Slf4j
@Component
public class RedisRunnerBean implements ApplicationRunner, CommandLineRunner, InitializingBean {

    /*
     * 参考文档：
     * https://github.com/spring-projects/spring-data-examples/tree/main/redis
     */

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private ThreadPoolTaskScheduler taskScheduler;
    @Resource
    private Client rebloomClient;
    @Resource
    private RedissonClient redisson;

    /**
     * 包括 Binary-safe strings
     * 包括 Bit arrays (or simply bitmaps)
     */
    private ValueOperations<String, Object> strings;
    private ListOperations<String, Object> lists;
    private SetOperations<String, Object> sets;
    private HashOperations<String, Object, Object> hashes;
    private ZSetOperations<String, Object> sortedSets;

    private HyperLogLogOperations<String, Object> hyperLogLogs;
    private GeoOperations<String, Object> geos;
    private StreamOperations<String, Object, Object> streams;

    @Resource
    private StreamMessageListenerContainer<String, MapRecord<String, String, String>> messageListenerContainer;

    /**
     * redis stream listener callback
     *
     * @param redisConnectionFactory org.springframework.data.redis.connection.RedisConnectionFactory
     * @return org.springframework.data.redis.stream.StreamMessageListenerContainer
     */
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> messageListenerContainer(RedisConnectionFactory redisConnectionFactory) {
        final StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions;
        containerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                .pollTimeout(Duration.ofMillis(1000L))
                .build();
        return StreamMessageListenerContainer.create(redisConnectionFactory, containerOptions);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 应用场景：商品编号、订单号采用INCR命令生成；文章阅读量；喜欢作者
        this.strings = this.redisTemplate.opsForValue();

        // 应用场景：微信订阅号的订阅文章列表
        this.lists = this.redisTemplate.opsForList();

        // 应用场景：抽奖程序的用户列表；微信朋友圈点赞列表；微博好友关注社交关系；QQ内推可能认识的人
        this.sets = this.redisTemplate.opsForSet();

        // 应用场景：低QPS下的商城购物车
        this.hashes = this.redisTemplate.opsForHash();

        // 应用场景：根据商品销售对商品进行排序显示；抖音热搜
        this.sortedSets = this.redisTemplate.opsForZSet();

        // bitmaps : 应用场景 ： 签到，京东签到领取京豆

        // 应用场景：天猫网站首页亿级UV的Redis统计方案
        this.hyperLogLogs = this.redisTemplate.opsForHyperLogLog();

        // 应用场景：美团地图位置附近的酒店推送
        this.geos = this.redisTemplate.opsForGeo();

        this.streams = this.redisTemplate.opsForStream();
    }

    /**
     * string数据结构
     */
    public void strings() {
        // SET
        this.strings.set("number", "123");

        // GET
        final Object number = this.strings.get("number");

        // MSET
        this.strings.multiSet(ImmutableMap.of("a1", "123", "a2", "234", "a3", "345"));

        // MGET
        final List<Object> multiGet = this.strings.multiGet(Arrays.asList("a1", "a2", "a3"));
        final String s = StringUtils.collectionToCommaDelimitedString(multiGet);

        // INCR (场景： 点赞数、文章阅读量)
        final Long aa1234 = this.strings.increment("aa1234");
        // INCRBY
        final Long incrBy12L = this.strings.increment("aa1234", 12L);

        // DECR
        final Long bb1234 = this.strings.decrement("bb1234");
        // DECRBY
        final Long decrBy12L = this.strings.decrement("bb1234", 12L);

        // STRLEN
        final Long size = this.strings.size("bb1234");

        // SETNX (分布式锁)
        final Boolean ifAbsent = this.strings.setIfAbsent("key", "value", 30L, TimeUnit.SECONDS);

        // SETEX
        this.strings.set("ex", "ex", 10L, TimeUnit.SECONDS);

        // GETSET
        final Object getset = this.strings.getAndSet("qwe", "新值");

        System.out.println("break-points");
    }

    /**
     * hash数据结构
     */
    public void hashes() {
        // HSET
        this.hashes.put("hashes-key", "username", "张三");
        this.hashes.put("hashes-key", "sex", "男");
        this.hashes.put("hashes-key", "age", "18");

        // HGET
        final Object age = this.hashes.get("hashes-key", "age");

        // HMSET
        this.hashes.putAll("hashes-hmset", ImmutableMap.of("username", "李四", "age", "28", "birthday", "2021-09-06"));

        // HMGET
        final List<Object> multiGet = this.hashes.multiGet("hashes-hmset", Arrays.asList("username", "age"));

        // HGETALL
        final Map<Object, Object> entries = this.hashes.entries("hashes-hmset");
        for (Map.Entry<Object, Object> entry : entries.entrySet()) {
            log.info("{}  -- {}", entry.getKey(), entry.getValue());
        }

        // HLEN
        final Long size = this.hashes.size("hashes-hmset");

        // HDEL
        final Long aLong = this.hashes.delete("hashes-hmset", "username");

        System.out.println("break-points");
    }

    /**
     * list数据结构
     */
    public void lists() {
        // LPUSH 向列表左边添加元素    (场景：微信公众号消息列表，最新消息会被lpush进来) (商品评论列表，按时间倒叙查看)
        final Long leftPush = this.lists.leftPush("lists-key", "向左插入");

        // RPUSH 向列表右边添加元素
        final Long rightPush = this.lists.rightPush("lists-key", "向右插入");

        // LRANGE 查看列表
        final List<Object> objectList = this.lists.range("lists-key", 0L, 1L);

        // LLEN 获取列表中元素的个数
        final Long aLong = this.lists.size("lists-key");

        System.out.println("break-points");
    }

    /**
     * set数据结构
     * 一个双端链表的结构，容量是2的32次方减1个元素，大概40多亿，主要功能有push/pop等，一般用在栈、队列、消息队列等场景。
     */
    public void sets() {
        // 抽奖小程序
        // sAdd 添加抽奖用户)    (sCard 统计多少用户参与)    (sRandMember 随机抽N个人，元素不删除)    (sPop 随机抽N个人，元素会删除

        // 朋友圈点赞列表
        // sAdd 点赞)    (sRem 取消点赞)    (sMembers 点赞用户列表)    (sCard 点赞用户数统计)    (sIsMember 判断某用户是否对当前用户点赞

        // 交集、并集、差集 可以用来做 共同关注的人、或者认识可能认识的人 等等

        // sAdd
        final Long aLong = this.sets.add("sets-keys", "张三", "李四", "王五", "赵六", "孙七", "周八", "吴九", "郑十");
        final Long aLongEng = this.sets.add("sets-keys-eng", "zhangsan", "lisi", "wangwu", "zhaoliu", "sunqi", "zhouba", "wujiu", "zhengshi");

        // sRem
        final Long remove = this.sets.remove("sets-keys", "王五", "孙七");

        // sMembers
        final Set<Object> members = this.sets.members("sets-keys");

        // sIsMember
        final Boolean isMember = this.sets.isMember("sets-keys", "吴九");

        // sCard
        final Long size = this.sets.size("sets-keys");

        // sRandMember
        final Object randomMember = this.sets.randomMember("sets-keys");

        // sPop
        final Object pop = this.sets.pop("sets-keys");

        // sDiff 差集
        final Set<Object> difference = this.sets.difference("sets-keys", "sets-keys-eng");

        // sInter 交集
        final Set<Object> intersect = this.sets.intersect("sets-keys", "sets-keys-eng");

        // sUnion 并集
        final Set<Object> union = this.sets.union("sets-keys", "sets-keys-eng");

        System.out.println("break-points");
    }

    /**
     * zset 有序列表
     */
    public void sortedSets() {
        // 应用场景：热搜排行榜；
        // 应用场景：淘宝商品按销量排序

        // zAdd
        final Boolean add0 = this.sortedSets.add("my-zset-key", "张三", 100D);
        final Boolean add1 = this.sortedSets.add("my-zset-key", "李四", 88D);
        final Boolean add2 = this.sortedSets.add("my-zset-key", "王五", 345D);
        final Boolean add3 = this.sortedSets.add("my-zset-key", "赵六", 220D);
        final Boolean add4 = this.sortedSets.add("my-zset-key", "孙七", 45D);
        final Boolean add5 = this.sortedSets.add("my-zset-key", "周八", 128D);

        // zRange 获取匹配下标的元素，按照分数从小到大排序
        final Set<Object> range = this.sortedSets.range("my-zset-key", 1, 3L);
        // zRevRange
        final Set<Object> reverseRange = this.sortedSets.reverseRange("my-zset-key", 1, 3L);

        // zScore 获取元素的分数
        final Double score = this.sortedSets.score("my-zset-key", "李四");

        // zRem 删除元素
        final Long remove = this.sortedSets.remove("my-zset-key", "王五");

        // zRangeByScore 获取指定分数范围的元素
        final Set<Object> rangeByScore = this.sortedSets.rangeByScore("my-zset-key", 80D, 120D);
        final Set<Object> reverseRangeByScore = this.sortedSets.reverseRangeByScore("my-zset-key", 80D, 120D);

        // zIncrBy 增加某个元素的分数
        this.sortedSets.incrementScore("my-zset-key", "李四", 30D);

        // zCard 获取集合中元素的数量
        final Long size = this.sortedSets.size("my-zset-key");

        // zCount 获得指定分数范围内的元素个数
        final Long count = this.sortedSets.count("my-zset-key", 80D, 120D);

        // zRemRangeByScore 按照排名范围删除元素
        final Long removeRangeByScore = this.sortedSets.removeRangeByScore("my-zset-key", 80D, 120D);

        // zRank 从小到大获取元素的排名
        final Long rank = this.sortedSets.rank("my-zset-key", "李四");

        // zRevRank 从大到小获取元素的排名
        final Long reverseRank = this.sortedSets.reverseRank("my-zset-key", "王五");

        System.out.println("break-points");
    }

    /**
     * bitmaps 用String类型作为底层数据结构实现的一种统计二值状态的数据类型
     * 由0和1状态表现的二进制位的bit数组(一个状态位占一个字节，一年才365字节)
     */
    public void bitmaps() {
        // 状态位的标记，true false
        // 用户是否登陆过Y、N，比如京东每日签到送京豆
        // 钉钉打卡上下班，签到统计

        this.strings.setBit("my-bitmaps-key", 0, true);
        this.strings.setBit("my-bitmaps-key", 1, false);
        this.strings.setBit("my-bitmaps-key", 2, true);
        this.strings.setBit("my-bitmaps-key", 3, true);

        this.strings.getBit("my-bitmaps-key", 0);

        // strLen
        this.strings.size("my-bitmaps-key");

        // bitCount
        final Long bigCount = this.redisTemplate.execute((RedisConnection conn) -> conn.bitCount("my-bitmaps-key".getBytes(StandardCharsets.UTF_8)));

        // bitOp
        final RedisCallback<Long> rc = (RedisConnection conn) -> {
            final byte[] destination = "dest-ack".getBytes(StandardCharsets.UTF_8);
            final byte[][] keys = {
                    "aa".getBytes(StandardCharsets.UTF_8)
                    , "bb".getBytes(StandardCharsets.UTF_8)
            };
            return conn.bitOp(RedisStringCommands.BitOperation.AND, destination, keys);
        };
        // 模拟连续两天签到数
        final Long bitOp = this.redisTemplate.execute(rc);

        this.signIn();

        System.out.println("break-points");
    }

    /**
     * 模拟签到
     */
    private void signIn() {
        final String bitMapKey = "user:zhang-san";

        // 执行签到
        for (int i = 0; i < 1024; i++) {
            if (i % 64 != 0) {
                this.strings.setBit(bitMapKey, i, Boolean.TRUE);
            }
        }
        // 获取某天签到情况
        final Boolean getBit = this.strings.getBit(bitMapKey, 365);
        log.info("第{}天{}签到！", 365, BooleanUtils.isTrue(getBit) ? "已" : "未");

        // 获取一共签到多少天
        final Long getBitCount = this.redisTemplate.execute((RedisConnection conn) -> conn.bitCount(bitMapKey.getBytes()));
        log.info("一共签到的天数：{}", getBitCount);

        // 判断用户连续签到数量
        // http://www.redis.cn/commands/bitfield.html
        final int limit = 365;
        final long offset = 1L;
        final BitFieldSubCommands bitFieldSubCommands = BitFieldSubCommands.create()
                .get(BitFieldSubCommands.BitFieldType.unsigned(limit))
                .valueAt(offset);
        // https://www.codenong.com/s1190000022557614/
        // https://www.alibabacloud.com/blog/a-small-victory-acceleration-of-bitfield-commands-for-apsaradb-for-redis_596774
        // 因为连接的是从节点，从节点只有读的权限，没有写的权限
        // READONLY You can't write against a read only replica.. channel:
//        final List<Long> longs = this.strings.bitField(bitMapKey, bitFieldSubCommands);
//        log.info("-----------{}", longs);
    }

    /**
     * 去重复统计功能的基数估计算法-就是HyperLogLog
     * 其实也是string类型
     */
    public void hyperLogLogs() {
        // 统计某个网站的UV、统计某个文章的UV
        // 用户搜索网站关键词的数量
        // 统计用户每天搜索不同词条个数

        // pfAdd 将所有元素添加到key中
        this.hyperLogLogs.add("my-hyperloglogs-keys11", "张三", "李四", "王五", "赵六");
        this.hyperLogLogs.add("my-hyperloglogs-keys22", "zhangsan", "lisi", "wangwu", "zhaoliu");

        // pfCount 统计key的估算值（不精确）
        final Long size = this.hyperLogLogs.size("my-hyperloglogs-keys11", "my-hyperloglogs-keys22");

        // pfMerge 合并key到新key
        final Long union = this.hyperLogLogs.union("my-hyperloglogs-newkey", "my-hyperloglogs-keys11", "my-hyperloglogs-keys22");

        System.out.println("break-points");
    }

    /**
     * geos类型
     * 其实底层是zset类型
     */
    public void geos() {
        // 美团、饿了么：
        // 当前位置商家推送（美食，娱乐，酒店等等）
        // 当前位置，附近共享单车、附近网约车

        // geoAdd
        this.geos.add("my-geos-key", new Point(116.393822, 39.90517), "人民大会堂");
        this.geos.add("my-geos-key", ImmutableMap.<Object, Point>builder()
                .put("广州塔", new Point(113.324463, 23.10647))
                .put("广州大剧院", new Point(113.322708, 23.114922))
                .put("广州周大福金融中心", new Point(113.325867, 23.117691))
                .put("华光庙", new Point(113.320518, 23.011831))
                .put("礼村小学", new Point(113.319374, 23.0135))
                .put("新月明珠花园", new Point(113.319832, 23.016251))
                .put("敏捷四季花园", new Point(113.317619, 23.015881))
                .build());

        // geoPos
        final List<Point> pointList = this.geos.position("my-geos-key", "人民大会堂");
        System.out.println(StringUtils.collectionToCommaDelimitedString(pointList));

        // geoDist
        final Distance distance = this.geos.distance("my-geos-key", "人民大会堂", "广州大剧院");
        System.out.println(distance);

        // geoRadiusByMember 计算1km范围内的商家
        final GeoResults<RedisGeoCommands.GeoLocation<Object>> results = this.geos.radius("my-geos-key", "礼村小学", new Distance(1D, Metrics.KILOMETERS));
        if (Objects.nonNull(results)) {
            for (GeoResult<RedisGeoCommands.GeoLocation<Object>> result : results) {
                final RedisGeoCommands.GeoLocation<Object> resultContent = result.getContent();
                final Distance resultDistance = result.getDistance();
                log.info("{}    {}", resultContent, resultDistance);
            }
        }

        // geoRadius 通过地点查找
        final RedisGeoCommands.GeoRadiusCommandArgs memberArgs = RedisGeoCommands.GeoRadiusCommandArgs
                .newGeoRadiusArgs()
                .includeDistance()
                .includeCoordinates()
                .sortAscending()
                .limit(15);
        this.geos.radius("my-geos-key", "人民大会堂", new Distance(1D, Metrics.KILOMETERS), memberArgs);

        // geoRadius 通过坐标查找
        final Circle within = new Circle(new Point(113.317619, 23.015881), new Distance(3D, Metrics.KILOMETERS));
        final RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs
                .newGeoRadiusArgs()
                .includeDistance()
                .includeCoordinates()
                .sortAscending()
                .limit(15L);
        final GeoResults<RedisGeoCommands.GeoLocation<Object>> geoResults = this.geos.radius("my-geos-key", within, args);
        System.out.println(geoResults);

        // geoHash base32编码
        final List<String> hash = this.geos.hash("my-geos-key", "新月明珠花园", "礼村小学", "华光庙");
        System.out.println(StringUtils.collectionToCommaDelimitedString(hash));

        System.out.println("break-points");
    }

    /**
     * @see <a href="https://docs.spring.io/spring-data/data-redis/docs/current/reference/html/#redis.streams">Redis Streams<a/>
     */
    public void streams() {
        // 生产者，每隔n秒执行一次
        this.taskScheduler.scheduleWithFixedDelay(() -> {

            final Map<String, String> raw = Collections.singletonMap(RandomStringUtils.randomAlphanumeric(13)
                    , RandomStringUtils.randomNumeric(9));
            final StringRecord record = StreamRecords.string(raw)
                    .withStreamKey("my-streams-key");
            final RecordId recordId = this.streams.add(record);
            System.out.println("provider: >>>>>    " + recordId);

        }, Duration.ofSeconds(10L).toMillis());

        // 消费者
        this.messageListenerContainer.receive(StreamOffset.latest("my-streams-key"), message -> {
            System.out.println("MessageId: " + message.getId());
            System.out.println("Stream: " + message.getStream());
            System.out.println("Body: " + message.getValue());
        });
    }

    /**
     * 由一个初值都为零的bit数组和多个哈希函数构成，用来快速判断某个数据是否存在
     * <p/>布隆过滤器是一种类似set的数据结构，只是统计结果不太准确
     * <p/>1、高效地插入和查询，占用空间少，返回的结果是不确定性的。
     * <p/>2、一个元素如果判断结果为存在的时候元素不一定存在，但是判断结果为不存在的时候则一定不存在。
     * <p/>3、布隆过滤器可以添加元素，但是不能删除元素。因为删掉元素会导致误判率增加。
     * <p/>4、误判只会发生在过滤器没有添加过的元素，对于添加过的元素不会发生误判。
     * <p>
     * <p/>结论备注：有，是可能有；无，是肯定无（可以保证的是，如果布隆过滤器判断一个元素不在一个集合中，那这个元素一定不会在集合中）
     */
    @SuppressWarnings({"unused"})
    public void bloomFilter() {
        // 场景：解决缓存穿透
        // 场景：黑名单策略，把非法的名单加入到bloomfilter，让请求先过bloomfilter
        // 场景：防止重复推送：视频、优惠券、等等

        // 布隆过滤器里预计要插入多少数据
        final int expectedInsertions = 1000 * 1000;
        // 误判率,它越小误判的个数也就越少(思考，是不是可以设置的无限小，没有误判岂不更好)
        final double fpp = 0.03D;
        this.bloomClientGuava(expectedInsertions, fpp);

        // -------------------------------------------------------------------------------------------------------------
        // redis RedisBloom + java client JReBloom
        final String name = "my-bloom-filter-key:phone";
        if (Boolean.FALSE.equals(this.redisTemplate.hasKey(name))) {
            this.rebloomClient.createFilter(name, expectedInsertions, fpp);
        }

        // 加载bloom-filter数据
        this.rebloomClient.add(name, "10086");
        this.rebloomClient.add(name, "13800138000");
        this.rebloomClient.add(name, "13005705672");

        // 假设这是redis真实数据，用bloomfilter解决缓存穿透，先去bloomfilter查询是否有此数据，再走正常业务流程，查redis查db
        this.redisson.getBucket("10086", StringCodec.INSTANCE).set("china-mobile-10086");
        this.redisson.getBucket("13800138000", StringCodec.INSTANCE).set("china-mobile-13800138000");
        this.redisson.getBucket("13005705672", StringCodec.INSTANCE).set("china-mobile-13005705672");
        this.redisson.getBucket("13556246780", StringCodec.INSTANCE).set("china-mobile-13556246780");

        // 挡住无效请求
        final boolean exists = this.rebloomClient.exists(name, "13794721175");
        System.out.println("redis bloom filter exists :" + exists);

        // 先去布隆过滤器里面查询
        if (this.rebloomClient.exists(name, "13800138000")) {
            // 有效请求，放行查询redis缓存
            final Object redisCacheValue = this.redisson.getBucket("13800138000", StringCodec.INSTANCE).get();
            if (Objects.nonNull(redisCacheValue)) {
                System.out.println("来自redis的缓存值：" + redisCacheValue);
            } else {
                // 取数据库查询值，查询到值，并缓存到redis
                // db : select * from xxx where id = "13800138000"
                // redis : set 13800138000 *
                System.out.println("来自db的缓存值：db-china-mobile-13800138000");
            }
        }
    }

    @SuppressWarnings({"UnstableApiUsage", "SameParameterValue"})
    private void bloomClientGuava(int expectedInsertions, double fpp) {
        // guava 创建布隆过滤器对象    （只能单机使用？）

        final BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), expectedInsertions, fpp);

        System.out.println(bloomFilter.mightContain(1));
        System.out.println(bloomFilter.mightContain(2));

        // 判断指定元素是否存在
        for (int insertions = expectedInsertions; insertions > 0; insertions--) {
            // 将元素添加进布隆过滤器
            bloomFilter.put(insertions);
        }

        // 故意取10万个不在过滤器里的值，看看有多少个会被认为在过滤器里
        final List<Integer> list = Lists.newArrayListWithExpectedSize(4096);
        for (int insertions = expectedInsertions + 100_000; insertions > expectedInsertions; insertions--) {
            if (bloomFilter.mightContain(insertions)) {
                System.out.println(insertions + "    被误判了！");
                list.add(insertions);
            }
        }
        // 100w的数据，传入10w个不相关的数据误判3033个，3033÷10w=0.03033
        System.out.println("误判的数量：" + list.size());
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
//        this.strings();
//        this.hashes();
//        this.lists();
//        this.sets();
//        this.sortedSets();
//
//        this.bitmaps();
//        this.hyperLogLogs();
//        this.geos();
//        // pub/sub 功能
//        this.streams();
//
//        // module bloom
//        this.bloomFilter();
    }

    @Override
    public void run(String... args) throws Exception {
    }

}
